package com.deepglint.window;

import com.deepglint.beans.SensorReading;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.assigners.TumblingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.apache.flink.util.IterableUtils;
import org.apache.flink.util.OutputTag;

/**
 * @author mj
 * @version 1.0
 * @date 2021-11-17 23:19
 */
public class WindowTest_WindowFunction {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        // 如果没有设置共享组，则复用前面的共享组，如果前面没有slot共享组，则有一个"default"共享组
        DataStream<String> streamSource = env.socketTextStream("192.168.150.128", 7777);

        DataStream<SensorReading> streamMap = streamSource.map((MapFunction<String, SensorReading>) value -> {
            String[] split = value.split(",");
            return new SensorReading(split[0], split[1], new Long(split[2]), new Double(split[3]));
        });

        streamMap.print();

        // 1、增量聚合函数
        DataStream<Integer> aggregateStream = streamMap.keyBy("id")
                .window(TumblingProcessingTimeWindows.of(Time.seconds(15)))
                .aggregate(new AggregateFunction<SensorReading, Integer, Integer>() {
                    @Override
                    public Integer createAccumulator() {
                        return 0;
                    }

                    @Override
                    public Integer add(SensorReading value, Integer accumulator) {
                        return accumulator + 1;
                    }

                    @Override
                    public Integer getResult(Integer accumulator) {
                        return accumulator;
                    }

                    @Override
                    public Integer merge(Integer a, Integer b) {
                        return a + b;
                    }
                });// 累加器

//        aggregateStream.print();

        // 全窗口计函数
        SingleOutputStreamOperator<Tuple3<String, Long, Integer>> resultSteam2 = streamMap.keyBy("id")
                .window(TumblingProcessingTimeWindows.of(Time.seconds(15)))
//                .process(new ProcessWindowFunction<SensorReading, Integer, Tuple, TimeWindow>() {
//                    @Override
//                    public void process(Tuple tuple, ProcessWindowFunction<SensorReading, Integer, Tuple, TimeWindow>.Context context, Iterable<SensorReading> elements, Collector<Integer> out) throws Exception {
//
//                    }
//                })
                .apply(new WindowFunction<SensorReading, Tuple3<String, Long, Integer>, Tuple, TimeWindow>() {
                    @Override
                    public void apply(Tuple tuple, TimeWindow window, Iterable<SensorReading> input, Collector<Tuple3<String, Long, Integer>> out) throws Exception {
                        String id = tuple.getField(0);
                        Long windowTime = window.getEnd();
                        Integer count = Math.toIntExact(IterableUtils.toStream(input).count());
                        out.collect(new Tuple3<>(id, windowTime, count));
                    }
                });

        // 3、其他可选API
        // 什么是迟到数据：这就涉及到时间语义了
        // 开始允许迟到选项，只针对EventTime
        OutputTag<SensorReading> outputTag = new OutputTag<>("later");
        SingleOutputStreamOperator<SensorReading> sumStream = streamMap.keyBy("id")
                .window(TumblingProcessingTimeWindows.of(Time.seconds(15)))
//                .trigger() // 触发器 定义 window 什么时候关闭，触发计算并输出结果
//                .evictor() // 移除器 定义移除某些数据的逻辑(类似于过滤器)
                .allowedLateness(Time.minutes(1)) // 允许处理迟到数据，到时间后窗口不关闭，等待一分钟
                .sideOutputLateData(outputTag) // 将所有迟到的数据放入侧边流
                .sum("temperature");

        // 拿取侧边流的数据
        DataStream<SensorReading> sideOutputStream = sumStream.getSideOutput(outputTag);

        // 4、完成的的调用流程
        // dataStream --keyBy----> KeyedStream ---window---> WindowedStream ---apply/aggregate等---> dataStream
        // dataStream --windowAll----> AllWindowStream ---sum等---> dataStream

        resultSteam2.print();

        env.execute();
    }
}

